package com.sanq.product.cab.lock.zookeeper;

import com.sanq.product.cab.locks.ILock;
import com.sanq.product.cab.locks.ILockCallback;
import com.sanq.product.cab.tools.log.LogUtil;
import com.sanq.product.cab.tools.strings.StringUtil;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.concurrent.CountDownLatch;

/**
 * com.sanq.product.cab.locks.zookeeper.ZookeeperLock
 *
 * @author sanq.Yan
 * @date 2020/3/1
 */
@Service("lock")
public class ZookeeperLock implements ILock, InitializingBean {

    private static final String ROOT_PATH = "zkLock";

    @Resource
    private CuratorFramework curatorFramework;
    private CountDownLatch countDownLatch = new CountDownLatch(1);

    /**
     * 得到Lock锁的key
     */
    @Override
    public String getLockKey(String key) {
        return "/" + ROOT_PATH + "/" + key;
    }

    @Override
    public String getLockValue() {
        return StringUtil.uuid();
    }

    /**
     * 加锁
     */
    @Override
    public void lock(String key, String value) {
        while (true) {
            try {
                curatorFramework
                        .create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(getLockKey(key));

                break;
            } catch (Exception e) {
                try {
                    if (countDownLatch.getCount() <= 0) {
                        countDownLatch = new CountDownLatch(1);
                    }
                    countDownLatch.await();
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }

    /**
     * 解锁
     */
    @Override
    public void unLock(String key, String value) {
        try {
            String path = getLockKey(key);
            if (curatorFramework.checkExists().forPath(path) != null) {
                curatorFramework.delete().forPath(path);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 另一种执行方式
     */
    @Override
    public void execute(String key, ILockCallback callback) {
        String path = getLockKey(key);

        String value = getLockValue();

        lock(path, value);
        try {
            callback.callback();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            unLock(path, value);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        curatorFramework = curatorFramework.usingNamespace("lock-namespace");

        try {
            String keyPath = "/" + ROOT_PATH;
            if (curatorFramework.checkExists().forPath(keyPath) == null) {
                curatorFramework.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(keyPath);
            }
            addWatcher(ROOT_PATH);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    private void addWatcher(String path) throws Exception {
        String keyPath;

        if (path.equals(ROOT_PATH)) {
            keyPath = "/" + path;
        } else {
            keyPath = getLockKey(path);
        }

        final PathChildrenCache cache = new PathChildrenCache(curatorFramework, keyPath, false);
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener((client, event) -> {
            if (cache.getListenable().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                String oldPath = event.getData().getPath();
                LogUtil.getInstance(getClass()).i("上一个节点{}已经断开", oldPath);
                if (oldPath.contains(path)) {
                    countDownLatch.countDown();
                }
            }
        });
    }
}
